package com.lyrcsoft.rokectmq.details;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

/**
 * @author fujc2dev@126.com
 * @date 2024-05-11 13:20
 */
public class ComuserDetails {

    public static void main(String[] args) throws Exception {

        //todo 属性
        // todo consumerGroup:消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("king");
        //todo 指定Namesrv地址信息.
        consumer.setNamesrvAddr("106.55.246.66:9876");
        //todo 消息消费艇式 (默认集群消费)
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //todo 指定消费开始偏移量(上次消费偏移量、最大偏移量、最小偏移量、启动时间戳) 开始消费
        // 启动时间戳 ：CONSUME_FROM_TIMESTAMP
        //  假设某一个群组是5.20号启动的，那我消费消息的时候，从5.20号启动的，那我消费的消息就是5.20号之后的消息，之前的消息都不管了
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //todo 消费长最小线程数付(默认20)
        consumer.setConsumeThreadMin(20);
        //todo 消费者最大线程数址(默认20)
        consumer.setConsumeThreadMax(20);
        //todo 推模式下任务间隔时间(推模式也是基于不断的轮训拉取的封装)
        consumer.setPullInterval(0);
        //todo 推模式下作务拉取的条数,默认32条(一批一批拉)
        consumer.setPullBatchSize(32);
        //todo 消息重试次数,-1代16次，超过限定次数会成为死信消息
        consumer.setMaxReconsumeTimes(-1);
        //todo 消息消费超时时间(消息可能阻塞正在使用的线程的最大时间：以分钟为单位)
        consumer.setConsumeTimeout(15);
        // todo 获取消费者对主题分配了那些消息队列
        Set<MessageQueue> MessageQueueSet = consumer.fetchSubscribeMessageQueues("TopicTest");
        Iterator iterator = MessageQueueSet.iterator();
        while (iterator.hasNext()) {
            MessageQueue MessageQueue = (MessageQueue)
                    iterator.next();
            System.out.println(MessageQueue.getQueueId());
        }
        //todo 方法订阅
        //todo 基于主题订阅消息，消息过滤使用表达式
        consumer.subscribe("TopicTest", "*");//tag tagA|TagB|TagC
        //todo
        consumer.subscribe("TopicTest", MessageSelector.bySql("a between @ and 3"));
        // todo
        consumer.subscribe("TopicTest", MessageSelector.byTag("tagA|TagB"));
        //todo 取消消息订阅
        consumer.unsubscribe("TopicTest");
        // todo 注册监听器
        // todo 注册并发事件监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        String topic = msg.getTopic();
                        String msgBody = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
                        String tags = msg.getTags();
                        System.out.println("收到消息：" + "topic：" + topic + "，tags：" + tags + "，body：" + msgBody);
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                        // 没有成功，回到重试队列中去
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // todo
        consumer.registerMessageListener(new MessageListenerOrderly() {

            /**
             * It is not recommend to throw exception,rather than returning ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT
             * if consumption failure
             *
             * @param msgs    msgs.size() >= 1<br> DefaultMQPushConsumer.consumeMessageBatchMaxSize=1,you can modify here
             * @param context
             * @return The consume status
             */
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        String topic = msg.getTopic();
                        String msgBody = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
                        String tags = msg.getTags();
                        System.out.println("收到消息：" + "topic：" + topic + "，tags：" + tags + "，body：" + msgBody);
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                        // 没有成功，回到重试队列中去
                        return ConsumeOrderlyStatus.ROLLBACK;
                    }
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
    }

}